pub mod auth;
use std::sync::Arc;

use chrono::{Duration, Utc};

use crm_metadata::pb::{Content, MaterializeRequest};
use crm_send::pb::SendRequest;
use futures::StreamExt;
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
use tonic::{Response, Status};
use user_stat::pb::QueryRequest;

use crate::{
    pb::{
        RecallRequest, RecallResponse, RemindRequest, RemindResponse, WelcomeRequest,
        WelcomeResponse,
    },
    CrmService,
};

impl CrmService {
    pub async fn welcome(
        &self,
        request: WelcomeRequest,
    ) -> Result<Response<WelcomeResponse>, Status> {
        let request_id = request.id;
        let d1 = Utc::now() - Duration::days(request.interval as _);
        let d2 = d1 + Duration::days(1);
        let query = QueryRequest::new_with_dt("created_at", d1, d2);
        let mut res_user_stats = self.user_stats.clone().query(query).await?.into_inner();
        let res_contents = self
            .metadata
            .clone()
            .materialize(MaterializeRequest::new_with_ids(&request.content_ids))
            .await?
            .into_inner();
        let contents: Vec<Content> = res_contents
            .filter_map(|v| async move { v.ok() })
            .collect()
            .await;
        let contents = Arc::new(contents);
        let (tx, rx) = mpsc::channel(1024);
        let sender_email = self.config.server.sender_email.clone();
        tokio::spawn(async move {
            while let Some(Ok(user)) = res_user_stats.next().await {
                let contents = contents.clone();
                let sender_email = sender_email.clone();
                let tx = tx.clone();
                let req = SendRequest::new(
                    "welcome ".to_string(),
                    sender_email,
                    &[user.email],
                    &contents,
                );
                tx.send(req).await.unwrap();
            }
        });
        let stream = ReceiverStream::new(rx);
        self.notification.clone().send(stream).await?;
        Ok(Response::new(WelcomeResponse { id: request_id }))
    }

    pub async fn recall(&self, request: RecallRequest) -> Result<Response<RecallResponse>, Status> {
        let request_id = request.id;
        let content_ids = request.content_ids;
        let laster_visit_interval = request.laster_visit_interval;
        let d1 = Utc::now();
        let d2 = d1 - Duration::days(laster_visit_interval as _);
        let query_request = QueryRequest::new_with_dt_and_ids(
            "last_visited_at",
            d2,
            d1,
            "rencet_watched",
            &content_ids,
        );
        let mut user_stats = self
            .user_stats
            .clone()
            .query(query_request)
            .await?
            .into_inner();
        let res_contents = self
            .metadata
            .clone()
            .materialize(MaterializeRequest::new_with_ids(&content_ids))
            .await?
            .into_inner()
            .filter_map(|v| async move { v.ok() })
            .collect::<Vec<Content>>()
            .await;
        let contents = Arc::new(res_contents);
        let (tx, rx) = mpsc::channel(1024);
        let sender_email = self.config.server.sender_email.clone();
        tokio::spawn(async move {
            while let Some(Ok(user)) = user_stats.next().await {
                let contents = contents.clone();
                let sender_email = sender_email.clone();
                let tx = tx.clone();
                let req = SendRequest::new(
                    "recall ".to_string(),
                    sender_email,
                    &[user.email],
                    &contents,
                );
                tx.send(req).await.unwrap();
            }
        });
        let stream = ReceiverStream::new(rx);
        self.notification.clone().send(stream).await?;
        Ok(Response::new(RecallResponse { id: request_id }))
    }

    pub async fn remind(&self, request: RemindRequest) -> Result<Response<RemindResponse>, Status> {
        let request_id = request.id;
        let d1 = Utc::now();
        let d2 = d1 - Duration::days(request.laster_visit_interval as _);
        let query_request: QueryRequest = QueryRequest::new_with_dt("last_visited_at", d2, d1);
        let mut user_stats = self
            .user_stats
            .clone()
            .query(query_request)
            .await?
            .into_inner();
        let (tx, rx) = mpsc::channel(1024);
        let sender_email = self.config.server.sender_email.clone();
        tokio::spawn(async move {
            while let Some(Ok(user)) = user_stats.next().await {
                let sender_email = sender_email.clone();
                let tx = tx.clone();
                let req = SendRequest::new("remind ".to_string(), sender_email, &[user.email], &[]);
                tx.send(req).await.unwrap();
            }
        });
        let stream = ReceiverStream::new(rx);
        self.notification.clone().send(stream).await?;
        Ok(Response::new(RemindResponse { id: request_id }))
    }
}
